package com.imooc.bilibili.websocket;

import com.alibaba.fastjson.JSONObject;
import com.imooc.bilibili.constant.MQConstant;
import com.imooc.bilibili.domain.Danmu;
import com.imooc.bilibili.service.DanmuService;
import com.imooc.bilibili.util.RocketMQUtil;
import com.imooc.bilibili.util.TokenUtil;
import com.mysql.cj.util.StringUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * WebSocket 相关的服务
 *
 * @author xiexu
 * @create 2022-07-07 09:26
 */
@Component
@ServerEndpoint("/imserver/{token}/{videoId}") // 表示这是一个跟WebSocket相关的服务类
public class WebSocketService {

    /**
     * 日志记录
     */
    private final Logger logger = LoggerFactory.getLogger(WebSocketService.class);

    /**
     * 一个客户端 关联 一个WebSocketService
     * 是一个多例模式的，这里需要注意下
     */
    public static final ConcurrentHashMap<String, WebSocketService> WEBSOCKET_MAP = new ConcurrentHashMap<>();

    /**
     * 服务端 和 客户端 进行通信的一个会话
     * 当我们有一个客户端进来了，然后保持长连接成功了，那我们就会保存一个跟这个客户端关联的session
     */
    private Session session;

    /**
     * 客户端的唯一标识
     */
    private String sessionId;

    /**
     * 用户id
     */
    private Long userId;

    /**
     * 视频id
     */
    private Long videoId;

    /**
     * 统计不同视频在线观看人数的key
     */
    private static final String VIDEO_ONLINE_NUMBER_KEY = "online-vdieo:";


    // 下面解释为什么不能这么用
//    @Autowired
//    private RedisTemplate redisTemplate;

    /**
     * 全局的上下文变量
     * 在多例模式下，springboot在第一个WebSocketService进行实现的时候，确实是会把RedisTemplate进行注入的，
     * 但是在往后的WebSocketService中就不会再注入RedisTemplate，因为springboot默认是单例的，它觉得注入一次之后就没必要再注入了，
     * 这就会引发问题，会导致往后的RedisTemplate都获取不到。
     * 解决方法：使用全局的上下文变量 ApplicationContext
     */
    private static ApplicationContext APPLICATION_CONTEXT;

    /**
     * 通用的上下文环境变量的方法，每个WebSocketService都会共用同一个ApplicationContext
     *
     * @param applicationContext
     */
    public static void setApplicationContext(ApplicationContext applicationContext) {
        WebSocketService.APPLICATION_CONTEXT = applicationContext;
    }

    /**
     * 打开连接
     *
     * @param session
     * @param token
     * @PathParam("token") 获取 @ServerEndpoint("/imserver/{token}") 后面的参数
     */
    @OnOpen // 连接成功后会自动调用该方法
    public void openConnection(Session session, @PathParam("token") String token, @PathParam("videoId") Long videoId) {
        // 如果是游客观看视频，虽然有弹幕，但是没有用户信息，所以需要用try
        try {
            this.userId = TokenUtil.verifyToken(token);
        } catch (Exception ignored) {
        }

        // 保存当前用户观看的视频id
        this.videoId = videoId;

        // 保存session相关信息到本地
        this.sessionId = session.getId();
        this.session = session;

        // 判断WEBSOCKET_MAP是否含有sessionId，有的话先删除再重新添加
        if (WEBSOCKET_MAP.containsKey(sessionId)) {
            WEBSOCKET_MAP.remove(sessionId);
            // this表示当前的WebSocketService类
            WEBSOCKET_MAP.put(sessionId, this);
        } else { // 没有的话就直接添加
            WEBSOCKET_MAP.put(sessionId, this);
        }
        // 打印日志
        logger.info("用户连接成功：" + sessionId);

        // 连接成功之后需要通知客户端，方便客户端进行后续操作
        try {
            this.sendMessage("0");
        } catch (Exception e) {
            logger.error("连接异常！");
        }
    }

    /**
     * 用户刷新页面或者关闭当前页面、服务端断开连接等操作，都需要关闭连接
     */
    @OnClose // 关闭连接后会自动调用该方法
    public void closeConnection() {
        if (WEBSOCKET_MAP.containsKey(sessionId)) {
            WEBSOCKET_MAP.remove(sessionId);
            // 打印日志
            logger.info("用户退出：" + sessionId);
        }
    }

    /**
     * 客户端（前端）发送消息给服务端
     *
     * @param message
     */
    @OnMessage // 当前端发送消息给服务端，会自动调用该方法
    public void onMessage(String message) {
        // 打印日志
        logger.info("用户信息：" + sessionId + "，报文：" + message);
        if (!StringUtils.isNullOrEmpty(message)) {
            try {
                // 群发消息（服务端拿到某一个客户端发来的弹幕信息之后，然后群发给所有正在与服务端连接的客户端）
                for (Map.Entry<String, WebSocketService> entry : WEBSOCKET_MAP.entrySet()) {
                    // 获取每一个与服务端连接的客户端的webSocketService服务
                    WebSocketService webSocketService = entry.getValue();

                    // 获取弹幕生产者
                    DefaultMQProducer danmusProducer = (DefaultMQProducer) APPLICATION_CONTEXT.getBean("danmusProducer");
                    JSONObject jsonObject = new JSONObject();
                    jsonObject.put("message", message);
                    jsonObject.put("sessionId", webSocketService.getSessionId());
                    Message msg = new Message(MQConstant.TOPIC_DANMUS, jsonObject.toJSONString().getBytes(StandardCharsets.UTF_8));
                    // 异步发送消息
                    RocketMQUtil.asyncSendMsg(danmusProducer, msg);
                }
                if (this.userId != null) {
                    // --------- 保存弹幕到数据库 ----------
                    // 将message转换成Danmu实体类的数据
                    Danmu danmu = JSONObject.parseObject(message, Danmu.class);
                    danmu.setUserId(userId);
                    danmu.setCreateTime(new Date());
                    DanmuService danmuService = (DanmuService) APPLICATION_CONTEXT.getBean("danmuService");
                    // 获取异步保存弹幕生产者
                    DefaultMQProducer asyncadddanmusProducer = (DefaultMQProducer) APPLICATION_CONTEXT.getBean("asyncadddanmusProducer");
                    // 将Danmu实体类发送到MQ中去
                    Message msg = new Message(MQConstant.TOPIC_ASYNCADDDANMUS, JSONObject.toJSONString(danmu).getBytes(StandardCharsets.UTF_8));
                    // 异步发送消息
                    RocketMQUtil.asyncSendMsg(asyncadddanmusProducer, msg);

                    // ----------- 保存弹幕到redis -----------
                    danmuService.addDanmusToRedis(danmu);
                }
            } catch (Exception e) {
                logger.error("弹幕接收出现问题！");
                e.printStackTrace();
            }
        }
    }

    /**
     * 发生错误之后的处理
     *
     * @param error
     */
    @OnError // 发生错误时会自动调用该方法
    public void onError(Throwable error) {

    }

    /**
     * 服务端发送消息给客户端
     *
     * @param message
     * @throws IOException
     */
    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }

    /**
     * 定时任务，每5秒统计一次各个视频的在线观看人数
     *
     * @throws IOException
     * @Scheduled(fixedRate = 5000) 标识该方法是一个定时任务，并且每隔5秒执行一次该方法
     */
    @Scheduled(fixedRate = 5000)
    private void noticeOnlineCount() throws IOException {
        for (Map.Entry<String, WebSocketService> entry : WebSocketService.WEBSOCKET_MAP.entrySet()) {
            WebSocketService webSocketService = entry.getValue();
            DanmuService danmuService = (DanmuService) APPLICATION_CONTEXT.getBean("danmuService");
            RedisTemplate redisTemplate = (RedisTemplate) APPLICATION_CONTEXT.getBean("redisTemplate");
            String key = VIDEO_ONLINE_NUMBER_KEY + videoId;

            // 判断会话是否还处于打开状态
            if (webSocketService.session.isOpen()) {
                // 统计各个视频的在线观看人数
                redisTemplate.opsForValue().increment(key);
            }

        }
    }

    public Session getSession() {
        return session;
    }

    public String getSessionId() {
        return sessionId;
    }

}
